package com.atguigu.dga.governance.assessor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.math.BigDecimal;
import java.net.URI;

import java.util.ArrayList;
import java.util.List;

import static java.security.AccessController.checkPermission;

@Component("FILE_ACCESS_PERMISSION")
public class FileAccessPermissionAssessor extends Assessor{
    @Override
    public void checkProblem(AssessParam assessParam, GovernanceAssessDetail governanceAssessDetail) throws  Exception {
        //    1 准备阶段
        //     1.1 递归的访问工具  FileSystem
        //     1.2  递归的起点   一级子目录集合
        //     1.3  结果容器   List<String>  问题文件（目录)的清单
        //     1.4  计算比较时用到的参数

        //1.1
        String tableFsPath = assessParam.getTableMetaInfo().getTableFsPath();
        String tableFsOwner = assessParam.getTableMetaInfo().getTableFsOwner();
        FileSystem fileSystem = FileSystem.get(new URI(tableFsPath), new Configuration(), tableFsOwner);
        //1.2
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path(tableFsPath));
        //  1.3  结果容器   List<String>  问题文件（目录)的清单
        List<String> beyondAccessList = new ArrayList<>();
        //     1.4  计算比较时用到的参数
        String metricParamsJson = assessParam.getGovernanceMetric().getMetricParamsJson();
        JSONObject paramJsonObj = JSON.parseObject(metricParamsJson);
        String filePermission = paramJsonObj.getString("file_permission");
        String dirPermission = paramJsonObj.getString("dir_permission");

        checkDirFileRec(fileSystem,fileStatuses,beyondAccessList,filePermission,dirPermission);

        //  3  判断给分
        //     List<String>  如果长度>0  ->  0分
        if(beyondAccessList.size()>0){
            governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
            governanceAssessDetail.setAssessProblem("文件或目录权限超过建议值："+  StringUtils.join(beyondAccessList,",") );
        }


    }


    //    2 执行阶段
    //     2.0  遍历一级节点集合
    //       2.1   目录
    //             1 计算比较-> 有问题的追加到 List<String>  2 展开下级节点集合 3 递归
    //        2.2   文件
    //              1 计算比较  -> 有问题的追加到 List<String>

    private  void checkDirFileRec(FileSystem fileSystem,
                                  FileStatus[] fileStatuses,
                                  List<String> beyondAccessList,
                                  String filePermission,
                                  String dirPermission) throws  Exception {
        for (FileStatus fileStatus : fileStatuses) {
            //       2.1   目录
            //             1 计算比较-> 有问题的追加到 List<String>  2 展开下级节点集合 3 递归
            if(fileStatus.isDirectory()){
                checkPermission(fileStatus,dirPermission,beyondAccessList);
                FileStatus[] subFileStatuses = fileSystem.listStatus(fileStatus.getPath());
                checkDirFileRec(fileSystem,subFileStatuses,beyondAccessList,filePermission,dirPermission);
            }else{
                //        2.2   文件
                //              1 计算比较  -> 有问题的追加到 List<String>
                checkPermission(fileStatus,filePermission,beyondAccessList);
            }
        }
    }

    private void checkPermission(FileStatus fileStatus, String permission, List<String> beyondAccessList) {
        int userOrdinal = fileStatus.getPermission().getUserAction().ordinal();
        int groupOrdinal = fileStatus.getPermission().getGroupAction().ordinal();
        int otherOrdinal = fileStatus.getPermission().getOtherAction().ordinal();

        Integer userPermission = Integer.valueOf(permission.substring(0, 1));
        Integer groupPermission = Integer.valueOf(permission.substring(1, 2));
        Integer otherPermission = Integer.valueOf(permission.substring(2));

        if (userOrdinal>userPermission
            || groupOrdinal>groupPermission
            || otherOrdinal>otherPermission){
            beyondAccessList.add(fileStatus.getPath().toString());
        }
    }
}
